package com.kafka.service;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

/**
 * 真实的Kafka生产者服务实现
 * 
 * <p>当配置kafka.enabled=true时激活，实际发送消息到Kafka集群</p>
 * 
 * <h3>配置要求：</h3>
 * <pre>
 * kafka:
 *   enabled: true
 *   bootstrap-servers: localhost:9092
 *   # 其他Kafka生产者配置
 * </pre>
 * 
 * <h3>使用示例：</h3>
 * <pre>
 * {@code
 * @Autowired
 * private KafkaProducerService producer;
 * 
 * // 异步发送
 * producer.sendMessageAsync("topic", "message");
 * 
 * // 同步发送
 * boolean success = producer.sendMessageSync("topic", "message");
 * }
 * </pre>
 * 
 * @see org.springframework.kafka.core.KafkaTemplate
 * @see NoOpKafkaProducerService
 */
@Service
@ConditionalOnProperty(name = "kafka.enabled", havingValue = "true", matchIfMissing = true)
public class RealKafkaProducerService implements KafkaProducerService {
    private static final Logger logger = LoggerFactory.getLogger(RealKafkaProducerService.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public RealKafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    public CompletableFuture<Void> sendMessageAsync(String topic, String message) {
        return kafkaTemplate.send(topic, message).thenAccept(result -> {
            RecordMetadata metadata = result.getRecordMetadata();
            logger.info("Message sent successfully - Topic: {}, Partition: {}, Offset: {}", 
                topic, metadata.partition(), metadata.offset());
        }).exceptionally(ex -> {
            logger.error("Failed to send message", ex);
            return null;
        });
    }

    @Override
    public boolean sendMessageSync(String topic, String message) {
        try {
            SendResult<String, String> result = kafkaTemplate.send(topic, message).get();
            logger.info("Message sent successfully - Topic: {}, Partition: {}, Offset: {}", 
                topic, result.getRecordMetadata().partition(), result.getRecordMetadata().offset());
            return true;
        } catch (Exception e) {
            logger.error("Failed to send message", e);
            return false;
        }
    }

    @Override
    public void sendMessageWithKey(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message).thenAccept(result -> {
            RecordMetadata metadata = result.getRecordMetadata();
            logger.info("Keyed message sent - Key: {}, Partition: {}, Offset: {}", 
                key, metadata.partition(), metadata.offset());
        }).exceptionally(ex -> {
            logger.error("Failed to send keyed message", ex);
            return null;
        });
    }
}